/************************************************************************
Copyright 2003, 2004, 2005, 2006, 2007, 2008 Grid Technology Research Center,
National Institute of Advanced Industrial Science and Technology.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
************************************************************************/

package jp.aist.gtrc.plus.scheduler.specific.pbs.status;

import java.util.Calendar;
import java.util.Collection;
import java.util.LinkedList;
import java.util.Locale;

import org.apache.log4j.Logger;

import jp.aist.gtrc.plus.reserve.ReserveId;
import jp.aist.gtrc.plus.scheduler.job.JobStateType;
import jp.aist.gtrc.plus.scheduler.node.ExecNodeAllocator;
import jp.aist.gtrc.plus.scheduler.node.NodeAllocateSet;
import jp.aist.gtrc.plus.scheduler.node.NodeResource;
import jp.aist.gtrc.plus.scheduler.status.ArchType;
import jp.aist.gtrc.plus.scheduler.status.JobID;
import jp.aist.gtrc.plus.scheduler.status.JobStatus;
import jp.aist.gtrc.plus.scheduler.status.JobStatusBase;
import jp.aist.gtrc.plus.scheduler.status.NodeStatus;
import jp.aist.gtrc.plus.scheduler.status.OSType;
import jp.aist.gtrc.plus.scheduler.status.QueueStatus;
import jp.aist.gtrc.plus.scheduler.util.TimeUtil;

//	attr			value
//	-----------------------------
//	14.koume.hpcc.jp	27
//		Job_Name		STDIN
//		Job_Owner		ohkubo@koume.hpcc.jp
//		job_state		Q
//		queue			workq
//		server			koume.hpcc.jp
//		Checkpoint		u
//		ctime			1124952952
//		Error_Path		koume.hpcc.jp:/home/ohkubo/STDIN.e14
//		Hold_Types		n
//		Join_Path		n
//		Keep_Files		n
//		Mail_Points		a
//		mtime			1124952952
//		Output_Path		koume.hpcc.jp:/home/ohkubo/STDIN.o14
//		Priority		0
//		qtime			1124952952
//		Rerunable		True
//		Resource_List	ncpus	1
//		Resource_List	nodect	1
//		Resource_List	walltime	00:15:00
//		substate		2
//		Variable_List	PBS_O_HOME=/home/ohkubo....
//		euser			ohkubo
//		egroup			ohkubo
//		queue_rank		5
//		queue_type		E
//		etime			1124952952
//		submit_args		-a 1800
//		x				rsvid:123


public class PBSJobStatus extends JobStatusBase implements StatusMap {

	static class RequestNodeSpec {

		private int nodeNum, procPerNode, cpuPerNode;

		RequestNodeSpec(String spec) {

			assert(spec != UNKNOWN_STATUS);
			nodeNum = 1;
			procPerNode = 1;
			cpuPerNode = 1;

                        //
                        // There is a lot of confusion among nodes,ppn,ncpus.
                        // The following explanation is not correct!
                        // Commented by Ekasit
			/*
			 * 'spec'				node	proc	cpu
			 * ------------------------------------------
			 * "x"					x		1		1
			 * "x:ppn=y"			x		y		1
			 * "x:ncpus=z"			x		1		z
			 * "x:ppn=y:ncpus=z"	x		y		z
			 * "ppn=y"				1		y		1
			 * "ncpus=z"			1		1		z
			 * "ppn=y:ncpus=z"		1		y		z
			 *
			 * x, y, z must be integer.
			 * ex: 'spec' = "1:ppn=3:ncpus=2" means
			 * run 3 process on a dual CPU node.
			 */
			String[] s = spec.split(":");
			for (int i = 0; i < s.length; i++) {
				String[] t = s[i].split("=");
				switch (t.length) {
				case 1:
					try {
						nodeNum = Integer.parseInt(s[i]);
					} catch (NumberFormatException e) {
						// TODO: maybe s[i] is hostname, not supported
						// ignore to continue
					}
					break;
				case 2:
					try {
						if (t[0].equals("ppn")) {
							procPerNode = Integer.parseInt(t[1]);
						} else if (t[0].equals("ncpus")) {
							cpuPerNode = Integer.parseInt(t[1]);
						} else {
							// invalid key is specified, ignore
						}
					} catch (NumberFormatException e) {
						// ignore to continue
					}
					break;
				default:
					break;
				}
			}

		}

		int getNodeNum() {
			return nodeNum;
		}

		int getProcPerNode() {
			return procPerNode;
		}

		int getCPUPerNode() {
			return cpuPerNode;
		}

	}

	private final StatusMap map;
	private QueueStatus queue;
	private boolean rsvIdChecked = false;
	private ReserveId rsvId;
	private LinkedList<RequestNodeSpec> reqNodeSpecs;
	private LinkedList<NodeAllocateSet> reqAllocSet;
	private OSType reqOS;
	private ArchType reqArch;

	private static final String RES_LIST = "Resource_List";
	private static final String RES_USED = "resources_used";
	private static final String RSVID_KEY = "rsvid";

	protected static final Logger logger = Logger.getLogger(PBSJobStatus.class);

	public PBSJobStatus(String name){

		map = new StatusMapImpl(name);

		//makeRequestNodeSpecs();
		//makeReserveId();

	}

	public String getQueueName() {

		return getStatus("queue");

	}

	public void setQueueStatus(QueueStatus queue) {

		/*
		 * set queue of this job is connected.
		 * use getQueueStatus().getPriority() to know queue priority.
		 */
		String qname = getQueueName();
		if (qname != UNKNOWN_STATUS) {
			if (qname.equals(queue.getName()) == false) {
				logger.debug("invalid queue name: " + queue.getName()
						+ ", must be " + qname);
				return;
			}
		}
		this.queue = queue;

	}

	public QueueStatus getQueueStatus() {

		return queue;

	}

	public JobID getJobID() {

		// "100.pbsserver.example.com" etc
		return new PBSJobID(map.getName());

	}

	public String getJobName(){

		// "STDIN" or user specified name
		return getStatus("Job_Name");

	}

	public String getOwner(){

		return getStatus("euser");

	}

	public String getOwnersGroup(){

		return getStatus("egroup");

	}

	public String getOwnersHost(){

		String[] s = getStatus("Job_Owner").split("@");
		if (s.length == 2)
			return s[1];
		else {
			logger.debug("Invalid Job_Owner: " + getStatus("Job_Owner"));
			return "";
		}

	}

	public String getJobState(){

		return getStatus("job_state");

	}

	public boolean isQueued(){

		return getJobState().equals("Q");

	}

	public boolean isRunning(){

		return getJobState().equals("R");

	}

        public boolean isCompleted() {
		return getJobState().equals("C");
        }

	public boolean isHeld() {
		return getJobState().equals("H");

/* Checking state H is more correct. - Ekasit
		// "n": not held, runnable if in queue
		// "u", "o", "s": held, not runnable
                String hold_types = getStatus("Hold_Types"); 
                if ( (hold_types == null) || hold_types.equals(UNKNOWN_STATUS) ) {
                      return false;
}
		return (getStatus("Hold_Types").equals("n") == false);
*/
	}

	public boolean isReadyToRun() {

		return (isQueued() && !isHeld());

	}

	public int getPriority(){

		String p = getStatus("Priority");
		if (p != UNKNOWN_STATUS)
			return Integer.parseInt(p);
		else
			return 0;

	}

	public void setPriority(int priority) {

		/*
		 * StarvedJobHelper changes job priority
		 * if it is starved.
		 */
		putStatus("Priority", Integer.toString(priority));

	}

	private void addRequestNodeSpec(RequestNodeSpec spec) {

		reqNodeSpecs.add(spec);
		NodeResource rsc = new NodeResource(
				spec.getCPUPerNode(), spec.getProcPerNode(),
				getRequestedArchType(), getRequestedOSType());
		NodeAllocateSet set = new NodeAllocateSet(spec.getNodeNum(), rsc);
		reqAllocSet.add(set);

	}

	private void makeRequestNodeSpecs() {

		reqNodeSpecs = new LinkedList<RequestNodeSpec>();
		reqAllocSet = new LinkedList<NodeAllocateSet>();

		String s = getStatus(RES_LIST, "nodes");
		if (s != UNKNOWN_STATUS) {
			for (String specTxt : s.split("\\+")) {
				addRequestNodeSpec(new RequestNodeSpec(specTxt));
			}
		} else {
			addRequestNodeSpec(new RequestNodeSpec("1"));
		}

	}

	public Collection<NodeAllocateSet> getNodeRequests() {

		if (reqAllocSet == null)
			makeRequestNodeSpecs();
		return reqAllocSet;

	}

        /*
         * Other resource requests use "-l other=..." as required properties of nodes
         */
        public String getOtherResourceRequests() {
		String s = getStatus(RES_LIST, "other");
		if (s == UNKNOWN_STATUS) {
			assert(false);
			s = null;
		}
		return s;
        }

	public String getRawNodeRequests() {

		String s = getStatus(RES_LIST, "nodes");
		if (s == UNKNOWN_STATUS) {
			assert(false);
			s = "1";
		}
		return s;

	}

        /*
         * return absolute deadline in time_t
         */
        public int getDeadline() {
                // s = "yyyy/MM/dd_HH:mm:ss" // obsoleted

                /* datetime in POSIX format
                 *
                 * [[[[CC]YY]MM]DD]hhmm[.SS]
                 *
                 * ex: "Thu Jun  1 16:27:50 JST 2006" is equal to
                 *              "200606011627.50"               (length = 15)
                 *                "0606011627.50"               (length = 13)
                 *                  "06011627.50"               (length = 11)
                 *                    "011627.50"               (length = 9)
                 *                      "1627.50"               (length = 7)
                 *
                 * ex: "Thu Jun  1 16:27:00 JST 2006" is equal to
                 *              "200606011627.00"  "200606011627"
                 *                "0606011627.00"    "0606011627"
                 *                  "06011627.00"      "06011627"
                 *                    "011627.00"        "011627"
                 *                      "1627.00"          "1627"
                 */


		String s = getStatus(RES_LIST, "deadline");
		if (s != UNKNOWN_STATUS) {
			try {
				return (int) (TimeUtil.makeCalendarFromPOSIX(s).getTimeInMillis() / 1000);
			} catch (NumberFormatException e) {
				// ignore, fall-through
			}
		}

                return Integer.MAX_VALUE;
        }

        /*
         * @return estimated job execution time (seconds) on the reference 
         *   machine obtained from the qsub's argument -W x=1000
         */
        public int getEstimatedExecutionTime() {
           String s = getStatus("x");
           if (s != UNKNOWN_STATUS) {
			try {
				return (int) Integer.parseInt(s);
			} catch (NumberFormatException e) {
				// ignore, fall-through
			}
           }
           return -1;
        }

        public boolean isQoSJob() {
           return (getDeadline() < Integer.MAX_VALUE);
        }

	public int getUsedCPUTime() {

		String s = getStatus(RES_USED, "cput");
		if (s != UNKNOWN_STATUS) {
			try {
				return TimeUtil.hhmmssToSec(s);
			} catch (NumberFormatException e) {
				// ignore, fall-through
			}
		}

		return 0;

	}

	public int getRequestedTime(){

		String s = getStatus(RES_LIST, "walltime");
		if (s != UNKNOWN_STATUS) {
			try {
				return TimeUtil.hhmmssToSec(s);
			} catch (NumberFormatException e) {
				// ignore, fall-through
			}
		}

		return WALLTIME_NOT_SPECIFIED;

	}

	private OSType getRequestedOSType() {

		if (reqOS != null)
			return reqOS;

		reqOS = OSType.Unknown;
		String s = getStatus(RES_LIST, "opsys");
		if (s != UNKNOWN_STATUS) {
			s = s.toLowerCase(Locale.US);
			if (s.indexOf("linux") != -1)
				reqOS = OSType.Linux;
			else if ((s.indexOf("solaris") != -1) || (s.indexOf("sunos") != -1))
				reqOS = OSType.Solaris;
		}

		return reqOS;

	}

	private ArchType getRequestedArchType() {

		if (reqArch != null)
			return reqArch;

		reqArch = ArchType.unknown;
		String s = getStatus(RES_LIST, "arch");
		if (s != UNKNOWN_STATUS) {
			s = s.toLowerCase(Locale.US);
			ArchType[] archs = ArchType.values();
			/*
			 * Search from tail.
			 * If 'arch' is "sparc64", we want to get sparc64, not sparc.
			 */
			for (int i = archs.length -1; i >= 0; i--) {
				ArchType arch = archs[i];
				if (s.indexOf(arch.name()) != -1) {
					reqArch = arch;
					break;
				}
			}
		}

		return reqArch;

	}

	public String getProperties() {

		return getStatus("properties");

	}

	public boolean isRunnableOn(NodeStatus node){

		ExecNodeAllocator allocator = new ExecNodeAllocator();
		return allocator.isReadyToRun((JobStatus)this, node);

	}

	public Calendar getSubmitTime() {

		String t = getStatus("ctime");
		if (t != UNKNOWN_STATUS) {
			return TimeUtil.makeCalendarFromSec(Integer.parseInt(t));
		} else {
			assert(false);
			return Calendar.getInstance();
		}

	}

	private String[] getSubmitArgs() {

		String s = getStatus("submit_args");
		if (s != UNKNOWN_STATUS) {
			return s.split("\\s");
		} else {
			return new String[]{};
		}

	}

	public Calendar getExecStartTime() {
		String t = getStatus("start_time");
		if (t != UNKNOWN_STATUS) {
			return TimeUtil.makeCalendarFromSec(Integer.parseInt(t));
		} else {
			assert(false);
			return Calendar.getInstance();
		}
	}

	public Calendar getRequestedExecStartTime() {

		String[] args = getSubmitArgs();
		for (int i = 0; i <= args.length -2; i++) {
			if (args[i].equals("-a")) {
				Calendar c = TimeUtil.makeCalendarFromPOSIX(args[i+1]);
				if (c != null) {
					return c;
				}
			}
		}
		return null;

	}

	private void makeReserveId() {

		/*
		 * To make reservation: way-1 (same as PBSPro)
		 *	> plus_reserve -s .....
		 *	... Reserve id is Rxxx
		 *
		 * Submit job to reserve queue, use -q option of qsub
		 * 	> qsub ... -q Rxxx
		 *
		 * This method returns "Rxxx" as ReserveId of the job.
		 */
		String qname = getQueueName();
		if (qname.charAt(0) == 'R') {
			rsvId = ReserveId.getInstance(qname);
			if (rsvId != null)
				return;
		}

		/*
		 * To make reservation: way-2 (same as Maui)
		 *	> plus_reserve -s .....
		 *	... Reserve id is Rxxx
		 *
		 * Submit job to reserve queue, use -x option of qsub
		 * 	> qsub ... -W x=rsvid:Rxxx
		 *
		 * This method returns "Rxxx" as ReserveId of the job.
		 */
		String s = getStatus("x");
		if (s != UNKNOWN_STATUS) {
			// s maybe "rsvid:Rxxx", Rxxx is ReserveId
			String[] v = s.split(":");
			if ((v.length == 2) && (v[0].equalsIgnoreCase(RSVID_KEY))) {
				rsvId = ReserveId.getInstance(v[1]);
			}
		}

	}

	public ReserveId getReserveId() {

		if (rsvIdChecked == false) {
			makeReserveId();
			rsvIdChecked = true;
		}
		return rsvId;

	}

	public JobStateType getState() {

		String s = getJobState();
		if (s.equals("Q"))
			return JobStateType.Queued;
		if (s.equals("R"))
			return JobStateType.Running;
		if (s.equals("E"))
			return JobStateType.Exiting;
		if (s.equals("H"))
			return JobStateType.Held;
		if (s.equals("S"))
			return JobStateType.Suspended;

		return JobStateType.Queued;

	}

	public StatusMap getStatusMap(String name) {
		return map.getStatusMap(name);
	}

	public void putStatus(String attr, String value) {
		map.putStatus(attr, value);
	}

	public String getStatus(String attr) {
		return map.getStatus(attr);
	}

	public String getStatus(String name, String attr) {
		return map.getStatus(name, attr);
	}

}
